
import Head from 'next/head'

<Head>
  <script>
    {
      `(function() {
         var _hmt = _hmt || [];
(function() {
  var hm = document.createElement("script");
  hm.src = "https://hm.baidu.com/hm.js?e60fb290e204e04c5cb6f79b0ac1e697";
  var s = document.getElementsByTagName("script")[0]; 
  s.parentNode.insertBefore(hm, s);
})();
       })();`
    }
  </script>
</Head>

![LangChain](https://pica.zhimg.com/50/v2-56e8bbb52aa271012541c1fe1ceb11a2_r.gif)





BabyAGI用户指南[#](#babyagi-user-guide "本标题永久链接")
===========


本文档演示了如何由[Yohei Nakajima](https://twitter.com/yoheinakajima)创建的[BabyAGI](https://github.com/yoheinakajima/babyagi/tree/main)。 BabyAGI是一个AI代理，可以基于给定的目标生成并假装执行任务。


本指南将帮助您了解创建您自己的递归代理程序的组件。


尽管BabyAGI使用特定的向量存储库/模型提供程序（Pinecone，OpenAI），之一实施它使用LangChain的好处是您可以轻松地为不同选项更换它们。在此实现中，我们使用FAISS向量存储库（因为它在本地运行并且免费）。



安装并导入所需模块[#](#install-and-import-required-modules "本标题永久链接")
-------------------------





```python
import os

from collections import deque

from typing import Dict, List, Optional, Any



from langchain import LLMChain, OpenAI, PromptTemplate

from langchain.embeddings import OpenAIEmbeddings

from langchain.llms import BaseLLM

from langchain.vectorstores.base import VectorStore

from pydantic import BaseModel, Field

from langchain.chains.base import Chain



```

连接向量存储库[#](#connect-to-the-vector-store "本标题永久链接")
-----------------------


根据您使用的向量存储库，此步骤可能会有所不同。





```python
from langchain.vectorstores import FAISS

from langchain.docstore import InMemoryDocstore



```


```python
# Define your embedding model

embeddings_model = OpenAIEmbeddings()

# Initialize the vectorstore as empty

import faiss



embedding_size = 1536

index = faiss.IndexFlatL2(embedding_size)

vectorstore = FAISS(embeddings_model.embed_query, index, InMemoryDocstore({}), {})



```

定义链 Define the Chains[#](#define-the-chains "Permalink to this headline")

-----------------


BabyAGI依赖于三个LLM链:


* 任务创建链，用于选择要添加到列表中的新任务
* 任务优先级链，用于重新设置任务的优先级
* 执行链，用于执行任务





```python
class TaskCreationChain(LLMChain):

 """Chain to generates tasks."""



    @classmethod

    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:

 """Get the response parser."""

        task_creation_template = (

            "You are a task creation AI that uses the result of an execution agent"

            " to create new tasks with the following objective: {objective},"

            " The last completed task has the result: {result}."

            " This result was based on this task description: {task_description}."

            " These are incomplete tasks: {incomplete_tasks}."

            " Based on the result, create new tasks to be completed"

            " by the AI system that do not overlap with incomplete tasks."

            " Return the tasks as an array."

        )

        prompt = PromptTemplate(

            template=task_creation_template,

            input_variables=[

                "result",

                "task_description",

                "incomplete_tasks",

                "objective",

            ],

        )

        return cls(prompt=prompt, llm=llm, verbose=verbose)



```


```python
class TaskPrioritizationChain(LLMChain):

 """Chain to prioritize tasks."""



    @classmethod

    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:

 """Get the response parser."""

        task_prioritization_template = (

            "You are a task prioritization AI tasked with cleaning the formatting of and reprioritizing"

            " the following tasks: {task_names}."

            " Consider the ultimate objective of your team: {objective}."

            " Do not remove any tasks. Return the result as a numbered list, like:"

            " #. First task"

            " #. Second task"

            " Start the task list with number {next_task_id}."

        )

        prompt = PromptTemplate(

            template=task_prioritization_template,

            input_variables=["task_names", "next_task_id", "objective"],

        )

        return cls(prompt=prompt, llm=llm, verbose=verbose)



```


```python
class ExecutionChain(LLMChain):

 """Chain to execute tasks."""



    @classmethod

    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:

 """Get the response parser."""

        execution_template = (

            "You are an AI who performs one task based on the following objective: {objective}."

            " Take into account these previously completed tasks: {context}."

            " Your task: {task}."

            " Response:"

        )

        prompt = PromptTemplate(

            template=execution_template,

            input_variables=["objective", "context", "task"],

        )

        return cls(prompt=prompt, llm=llm, verbose=verbose)



```
### 定义BabyAGI控制器[#](#define-the-babyagi-controller "Permalink to this headline")


BabyAGI将以上定义的链组合成（可能是）无限循环。





```python
def get_next_task(

    task_creation_chain: LLMChain,

    result: Dict,

    task_description: str,

    task_list: List[str],

    objective: str,

) -> List[Dict]:

 """Get the next task."""

    incomplete_tasks = ", ".join(task_list)

    response = task_creation_chain.run(

        result=result,

        task_description=task_description,

        incomplete_tasks=incomplete_tasks,

        objective=objective,

    )

    new_tasks = response.split("")

    return [{"task_name": task_name} for task_name in new_tasks if task_name.strip()]



```


```python
def prioritize_tasks(

    task_prioritization_chain: LLMChain,

    this_task_id: int,

    task_list: List[Dict],

    objective: str,

) -> List[Dict]:

 """Prioritize tasks."""

    task_names = [t["task_name"] for t in task_list]

    next_task_id = int(this_task_id) + 1

    response = task_prioritization_chain.run(

        task_names=task_names, next_task_id=next_task_id, objective=objective

    )

    new_tasks = response.split("")

    prioritized_task_list = []

    for task_string in new_tasks:

        if not task_string.strip():

            continue

        task_parts = task_string.strip().split(".", 1)

        if len(task_parts) == 2:

            task_id = task_parts[0].strip()

            task_name = task_parts[1].strip()

            prioritized_task_list.append({"task_id": task_id, "task_name": task_name})

    return prioritized_task_list



```


```python
def _get_top_tasks(vectorstore, query: str, k: int) -> List[str]:

 """Get the top k tasks based on the query."""

    results = vectorstore.similarity_search_with_score(query, k=k)

    if not results:

        return []

    sorted_results, _ = zip(*sorted(results, key=lambda x: x[1], reverse=True))

    return [str(item.metadata["task"]) for item in sorted_results]





def execute_task(

    vectorstore, execution_chain: LLMChain, objective: str, task: str, k: int = 5

) -> str:

 """Execute a task."""

    context = _get_top_tasks(vectorstore, query=objective, k=k)

    return execution_chain.run(objective=objective, context=context, task=task)



```


```python
class BabyAGI(Chain, BaseModel):

 """Controller model for the BabyAGI agent."""



    task_list: deque = Field(default_factory=deque)

    task_creation_chain: TaskCreationChain = Field(...)

    task_prioritization_chain: TaskPrioritizationChain = Field(...)

    execution_chain: ExecutionChain = Field(...)

    task_id_counter: int = Field(1)

    vectorstore: VectorStore = Field(init=False)

    max_iterations: Optional[int] = None



    class Config:

 """Configuration for this pydantic object."""



        arbitrary_types_allowed = True



    def add_task(self, task: Dict):

        self.task_list.append(task)



    def print_task_list(self):

        print("\033[95m\033[1m" + "*****TASK LIST*****" + "\033[0m\033[0m")

        for t in self.task_list:

            print(str(t["task_id"]) + ": " + t["task_name"])



    def print_next_task(self, task: Dict):

        print("\033[92m\033[1m" + "*****NEXT TASK*****" + "\033[0m\033[0m")

        print(str(task["task_id"]) + ": " + task["task_name"])



    def print_task_result(self, result: str):

        print("\033[93m\033[1m" + "*****TASK RESULT*****" + "\033[0m\033[0m")

        print(result)



    @property

    def input_keys(self) -> List[str]:

        return ["objective"]



    @property

    def output_keys(self) -> List[str]:

        return []



    def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:

 """Run the agent."""

        objective = inputs["objective"]

        first_task = inputs.get("first_task", "Make a todo list")

        self.add_task({"task_id": 1, "task_name": first_task})

        num_iters = 0

        while True:

            if self.task_list:

                self.print_task_list()



                # Step 1: Pull the first task

                task = self.task_list.popleft()

                self.print_next_task(task)



                # Step 2: Execute the task

                result = execute_task(

                    self.vectorstore, self.execution_chain, objective, task["task_name"]

                )

                this_task_id = int(task["task_id"])

                self.print_task_result(result)



                # Step 3: Store the result in Pinecone

                result_id = f"result_{task['task_id']}"

                self.vectorstore.add_texts(

                    texts=[result],

                    metadatas=[{"task": task["task_name"]}],

                    ids=[result_id],

                )



                # Step 4: Create new tasks and reprioritize task list

                new_tasks = get_next_task(

                    self.task_creation_chain,

                    result,

                    task["task_name"],

                    [t["task_name"] for t in self.task_list],

                    objective,

                )

                for new_task in new_tasks:

                    self.task_id_counter += 1

                    new_task.update({"task_id": self.task_id_counter})

                    self.add_task(new_task)

                self.task_list = deque(

                    prioritize_tasks(

                        self.task_prioritization_chain,

                        this_task_id,

                        list(self.task_list),

                        objective,

                    )

                )

            num_iters += 1

            if self.max_iterations is not None and num_iters == self.max_iterations:

                print(

                    "\033[91m\033[1m" + "*****TASK ENDING*****" + "\033[0m\033[0m"

                )

                break

        return {}



    @classmethod

    def from_llm(

        cls, llm: BaseLLM, vectorstore: VectorStore, verbose: bool = False, **kwargs

    ) -> "BabyAGI":

 """Initialize the BabyAGI Controller."""

        task_creation_chain = TaskCreationChain.from_llm(llm, verbose=verbose)

        task_prioritization_chain = TaskPrioritizationChain.from_llm(

            llm, verbose=verbose

        )

        execution_chain = ExecutionChain.from_llm(llm, verbose=verbose)

        return cls(

            task_creation_chain=task_creation_chain,

            task_prioritization_chain=task_prioritization_chain,

            execution_chain=execution_chain,

            vectorstore=vectorstore,

            **kwargs,

        )



```

### 运行BabyAGI[#](#run-the-babyagi "Permalink to this headline")


现在该创建BabyAGI控制器并观察其尝试完成您的目标了。





```python
OBJECTIVE = "Write a weather report for SF today"



```


```python
llm = OpenAI(temperature=0)



```


```python
# Logging of LLMChains

verbose = False

# If None, will keep on going forever

max_iterations: Optional[int] = 3

baby_agi = BabyAGI.from_llm(

    llm=llm, vectorstore=vectorstore, verbose=verbose, max_iterations=max_iterations

)



```



```python
baby_agi({"objective": OBJECTIVE})



```
```python


*****TASK LIST*****



1: Make a todo list



*****NEXT TASK*****



1: Make a todo list



*****TASK RESULT*****

1. Check the temperature range for the day.

2. Gather temperature data for SF today.

3. Analyze the temperature data and create a weather report.

4. Publish the weather report.



*****TASK LIST*****



2: Gather data on the expected temperature range for the day.

3: Collect data on the expected precipitation for the day.

4: Analyze the data and create a weather report.

5: Check the current weather conditions in SF.

6: Publish the weather report.



*****NEXT TASK*****



2: Gather data on the expected temperature range for the day.



*****TASK RESULT*****

I have gathered data on the expected temperature range for the day in San Francisco. The forecast is for temperatures to range from a low of 55 degrees Fahrenheit to a high of 68 degrees Fahrenheit.



*****TASK LIST*****



3: Check the current weather conditions in SF.

4: Calculate the average temperature for the day in San Francisco.

5: Determine the probability of precipitation for the day in San Francisco.

6: Identify any potential weather warnings or advisories for the day in San Francisco.

7: Research any historical weather patterns for the day in San Francisco.

8: Compare the expected temperature range to the historical average for the day in San Francisco.

9: Collect data on the expected precipitation for the day.

10: Analyze the data and create a weather report.

11: Publish the weather report.



*****NEXT TASK*****



3: Check the current weather conditions in SF.



*****TASK RESULT*****

I am checking the current weather conditions in SF. According to the data I have gathered, the temperature in SF today is currently around 65 degrees Fahrenheit with clear skies. The temperature range for the day is expected to be between 60 and 70 degrees Fahrenheit.



*****TASK ENDING*****





```




```python
{'objective': 'Write a weather report for SF today'}



```



